package com.doit.dw.dws

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}

import scala.collection.immutable
import scala.collection.mutable.ListBuffer

/**
 * @Date: 23.6.10 
 * @Author: Hang.Nian.YY
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object MakeAccPageWideJob {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    val session = SparkSession.builder()
      .master("local[*]")
      .appName(this.getClass.getSimpleName)
      .enableHiveSupport()
      .getOrCreate()
//----------------------------------加载hive数据------------------------------------------------------------------------------
   // 中心事实表  关联各维表  构建宽表  维度字段   指标字段
    //  只有follow_pv 这个字段没有处理
    val pageFrame = session.sql(
      """
        |with base_res as
        |    (select page_url,
        |            ref_url,
        |            guid,
        |            session_id,
        |            -- 0 as  follow_pv ,
        |            if(ts == max(ts) over (partition by session_id), 1, 0) as is_exit,
        |            if(ts == min(ts) over (partition by session_id), 1, 0) as is_enter,
        |            if(count(1) over (partition by session_id) == 1, 1, 0) as is_jump_out,
        |            -- 0 acc_long ,
        |            ref_type,
        |            -- 0 as  is_new ,
        |            provice       as province,
        |            city,
        |            device_type,
        |            os_name,
        |            release_channel,
        |            ts
        |     from doe39.dws_app_log_flow_page_dtl where dt = '2023-06-02') ,
        |user_info as (
        |   select
        |       guid , is_new
        |       from doe39.dws_app_log_flow_acount_agg
        |    where dt = '2023-06-02'
        |    ) ,
        |  session_agr as  (
        |        select
        |            session_id ,
        |            stop_ts
        |            from
        |                doe39.dws_app_log_flow_session_agg
        |            where  dt = '2023-06-02'
        |    )
        |select
        |    base_res.* ,
        |    user_info.is_new ,
        |    lead(base_res.ts ,1 , session_agr.stop_ts) over(partition by base_res.session_id order by base_res.ts) - base_res.ts  as acc_long
        |    from base_res join  user_info
        |on base_res.guid = user_info.guid
        |join
        |session_agr
        |on  session_agr.session_id = base_res.session_id
        |""".stripMargin)
//---------------------------------------------------------------------------------------------------------------
    /**
     * root
     * |-- page_url: string (nullable = true)
     * |-- ref_url: string (nullable = true)
     * |-- guid: long (nullable = true)
     * |-- session_id: string (nullable = true)
     * |-- is_exit: integer (nullable = false)
     * |-- is_enter: integer (nullable = false)
     * |-- is_jump_out: integer (nullable = false)
     * |-- ref_type: string (nullable = true)
     * |-- province: string (nullable = true)
     * |-- city: string (nullable = true)
     * |-- device_type: string (nullable = true)
     * |-- os_name: string (nullable = true)
     * |-- release_channel: string (nullable = true)
     * |-- ts: long (nullable = true)
     * |-- is_new: integer (nullable = true)
     * |-- acc_long: long (nullable = true)
     */
    /**
     * 第一步  解析数据  转换成RDD[PageBean]
     */
    /*  val  rdd  = pageFrame.rdd.map(row=>{
        // pageBeanfollow_pv属性
        PageBean(所有的属性 , follow_pv=0)
      })*/
    //---------------------------------------------------------------------------------------------------------
    // 解析数据  封装成 Bean
    val beans: RDD[PageBean] = pageFrame.rdd.map(row => {
      val page_url = row.getAs[String]("page_url")
      val ref_url = row.getAs[String]("ref_url")
      val guid = row.getAs[Long]("guid")
      val session_id = row.getAs[String]("session_id")
      val is_exit = row.getAs[Int]("is_exit")
      val is_enter = row.getAs[Int]("is_enter")
      val is_jump_out = row.getAs[Int]("is_jump_out")
      val ref_type = row.getAs[String]("ref_type")
      val province = row.getAs[String]("province")
      val city = row.getAs[String]("city")
      val device_type = row.getAs[String]("device_type")
      val os_name = row.getAs[String]("os_name")
      val release_channel = row.getAs[String]("release_channel")
      val ts = row.getAs[Long]("ts")
      val acc_long = row.getAs[Long]("acc_long")
      val is_new = row.getAs[Int]("is_new")
      PageBean(page_url,
        ref_url,
        guid,
        session_id,
        is_exit,
        is_enter,
        is_new,
        is_jump_out,
        ref_type,
        province,
        city,
        device_type,
        os_name,
        release_channel,
        acc_long,
        ts,
        0,
        "2023-06-02"
      )
    })
    //---------------------------------------------------------------------------------------------------------

    /**
     * 第二步  在会话内部 根据页面的访问关系  构建树
     * 深度遍历各个节点的下游节点数   --> 当前节点的 follow_pv=?
     * PageBean(一个页面 信息   , follow_pv)
     */
    val grouped = beans.groupBy(_.session_id)  //-------------------------------------------分组
    val roots: RDD[PageNode] = grouped.map(tp => {
      val session_id = tp._1
      val iters = tp._2 // 当前会话的 访问页面信息
      // 按照时间排序
      // 按照时间排序后的访问的页面明细    //------------------------------------------- 组内排序
      val sorted: immutable.Seq[PageBean] = iters.toList.sortBy(_.ts)
      var seqs: ListBuffer[PageNode] = null

                                       //-------------------------------------------构建树
      for (elem <- sorted) { // 遍历每条页面访问记录
        // 将每条页面访问记录  构建成节点
        val currentNode = PageNode(elem, ListBuffer[PageNode]())
        if (seqs == null) {
          seqs = new ListBuffer[PageNode]()
          seqs.append(currentNode)
        } else {
          seqs.append(currentNode)

          val maybeNode: Option[PageNode] = seqs.reverse.find(p => p.pageBean.page_url.equals(currentNode.pageBean.ref_url))
          if (maybeNode.isDefined) { // 找到了要挂载到的最近的节点
            maybeNode.get.chrldren.append(currentNode)
          }
        }
      }
      val root = seqs.head   //------------------------------------------- 只处理根节点
      root
    })

                              //------------------------------------------- 调用树遍历函数  处理跟节点
    val resBean: RDD[PageBean] = roots.flatMap(root => { // 一个会话中所有页面组成的根节点
      val midRes = ListBuffer[(PageBean, Int)]()
      followPvPlus(root, midRes)
      val pageBeans = midRes.map(tp => {
        tp._1.follow_pv = tp._2
        tp._1
      })
      pageBeans
    })
    //---------------------------------------------------------------------------------------------------------
    /**
     * 第三步  将结果保存到hive表中
     */
    import session.implicits._
    resBean.toDF.write.format("orc").mode(SaveMode.Overwrite).partitionBy("dt")
      .saveAsTable("doe39.dws_app_log_flow_acc_page_wide")


    //beans.groupby(session_id)
    /**
     * 第三步  将结果保存到hive表中
     */
    //rdd.toDF.write.saveAsTable()

    session.close()
  }

  def followPvPlus(node: PageNode, midRes: ListBuffer[(PageBean, Int)]): Int = {
    var pv: Int = 0

    val children = node.chrldren
    if (children != null && children.size > 0) {
      pv = children.size
      for (elem <- children) {
        var url = elem.pageBean
        pv += followPvPlus(elem, midRes)
      }
    }
    midRes.append((node.pageBean, pv))
    pv
  }

}
